/*
 * Hyuga: A Density-Grid Stream Clustering Platform.
 * Copyright (C) 2014 PUC-Rio/Laboratory for Advanced Collaboration
 *
 * Hyuga is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Hyuga is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Hyuga.  If not, see <http://www.gnu.org/licenses/>.
 */
package br.pucrio.inf.lac.konoha.hyuga.core;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.tuple.MutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.log4j.Logger;

import com.espertech.esper.client.Configuration;
import com.espertech.esper.client.ConfigurationEngineDefaults.Threading;
import com.espertech.esper.client.EPAdministrator;
import com.espertech.esper.client.EPPreparedStatement;
import com.espertech.esper.client.EPRuntime;
import com.espertech.esper.client.EPServiceProvider;
import com.espertech.esper.client.EPServiceProviderManager;
import com.espertech.esper.client.EPStatement;
import com.espertech.esper.client.UpdateListener;
import com.fasterxml.jackson.databind.ObjectMapper;

import br.pucrio.inf.lac.konoha.hyuga.common.Geo;
import br.pucrio.inf.lac.konoha.hyuga.common.Grid;
import br.pucrio.inf.lac.konoha.hyuga.datastructure.SegmentTree;
import br.pucrio.inf.lac.konoha.hyuga.events.CEPEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.CEPEvent.TYPE;
import br.pucrio.inf.lac.konoha.hyuga.events.LocationUpdateEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellChangedEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellContentEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellCoreEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellDisperseEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellRecheckEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellRecheckNeededEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellTransientCheckEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellTransientEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.csv.CSVLocationUpdateEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.grid.GridClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.grid.GridLogicalClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.grid.LockedCellClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.grid.LockedCellDisperseEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.grid.UnlockCellDisperseEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.semantics.DispersedGridClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.semantics.MergeGridClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.semantics.NewCellClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.semantics.NewCellDisperseEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.semantics.UpdateGridClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.listeners.CSVListener;
import br.pucrio.inf.lac.konoha.hyuga.listeners.CellOutputListener;
import br.pucrio.inf.lac.konoha.hyuga.listeners.ClusterDispersionListener;
import br.pucrio.inf.lac.konoha.hyuga.listeners.GridReadUnlockListener;
import br.pucrio.inf.lac.konoha.hyuga.listeners.GridWriteUnlockListener;
import br.pucrio.inf.lac.konoha.hyuga.listeners.GridLockListener;
import lac.cnet.sddl.objects.ApplicationObject;
import lac.cnet.sddl.objects.Message;
import lac.cnet.sddl.objects.SpatialMessage;
import lac.cnet.sddl.udi.core.listener.UDIDataReaderListener;

/**
 * This class implement the EPN (set of EPAs) that form the Hyuga algorithm. Read "An On-line Algorithm for Cluster
 * Detection of Mobile Nodes through Complex Event Processing" for further details.
 * 
 * Hyuga's EPAs can be deployed entirely in a single machine, or in a set of machines (distributed).
 * By default, EPAs output are JSON representations of the events, except for the final output.
 * Currently, we provide support for the ESPER CEP engine.
 * 
 * @author Marcos Roriz <mroriz@inf.puc-rio.br>
 */
public class HyugaService implements UDIDataReaderListener<ApplicationObject> {

    /** LOGGER. */
    private static Logger LOGGER = Logger.getLogger(HyugaService.class.getCanonicalName());

    /** DDSLink (Data Stream). */
    private DDSLink dataStream;
    
    /** Monitored domain (scenario). */
    private Scenario scenario;
    
    // Latitude and Longitude Partition Segments
    private SegmentTree latitudeSegments;
    private SegmentTree longitudeSegments;
    
    // Local Deploy
    private boolean localDeploy;

    /** Output Singleton Instance. */
    private Output output = Output.getInstance();
    
    /** Receiving Thread Pool. */
    private ExecutorService recvPool;

    /** JSON Mapper. */
    private ObjectMapper jsonMapper;

    /** CEP Engine. */
    private EPServiceProvider cepEngine;

    /** EPL Map. [Key, EPLContent, Listener] */
    private Map<EPA, Pair<EPPreparedStatement, Object>> epls;

    /** Deployed EPAs */
    private Map<EPA, EPStatement> deployedEPAs;
    
    /**
     * Hyuga Algorithm Constructor.
     * 
     * @param stream    a DDS data stream.
     * @param scenario  the {@link Scenario} configuration (latitude and longitude range, ε, minPts, etc)
     * @param debug     should we include debug rules?  
     */
    public HyugaService(DDSLink stream, Scenario scenario, boolean debug) {
        this.dataStream   = stream;
        this.scenario     = scenario;
        
        this.jsonMapper   = new ObjectMapper();
        this.recvPool     = Executors.newFixedThreadPool(16);
        this.epls         = new HashMap<EPA, Pair<EPPreparedStatement, Object>>();
        this.deployedEPAs = new HashMap<EPA, EPStatement>();
                
        // Initialize context partition segments
        preparePartitionMapper(scenario.getMinimumLatitude(),  scenario.getMaximumLatitude(),
                               scenario.getMinimumLongitude(), scenario.getMaximumLongitude(), scenario.getEps());

        // CEP rules and EPN
        prepareCEP(scenario, debug);
    }
    
    /**
     * Define the segments interval for the location update to context partition mapping function.
     * The segments are divided in the following intervals:
     *  [minLat, minLat + ε, minLat + 2ε, ..., maxLat] and 
     *  [minLng, minLng + ε, minLng + 2ε, ..., maxLng].
     * 
     * @param minLat the minimum latitude monitored.
     * @param maxLat the maximum latitude monitored.
     * @param minLng the minimum longitude monitored.
     * @param maxLng the maximum longitude monitored.
     * @param eps    maximum distance between nodes to form a cluster.
     */
    private void preparePartitionMapper(double minLat, double maxLat, double minLng, double maxLng, double eps) {
        ArrayList<Double> latitudeValues  = new ArrayList<Double>();
        ArrayList<Double> longitudeValues = new ArrayList<Double>();

        // Segments intervals, [minLat, minLat + ε, ..., maxLat] and [minLng, minLng + ε, ..., maxLng]
        double currentLat = minLat;
        double currentLng = minLng;
        do {
            if (currentLat < maxLat) latitudeValues.add(currentLat);
            if (currentLng < maxLng) longitudeValues.add(currentLng);

            currentLng = Geo.getOffsetLongitude(currentLat, currentLng, eps);
            currentLat = Geo.getOffsetLatitude(currentLat, eps);
        } while ((currentLat < maxLat) || (currentLng < maxLng));
        
        /* Add lower and higher bounds (-90 and 90 degrees for lat) and (-180 and 180 for longitude),
         * for values outside our monitoring space, in the first and last position of the intervals. */
        latitudeValues.add(0, (double) -90);
        latitudeValues.add((double) 90);

        longitudeValues.add(0, (double) -180);
        longitudeValues.add((double) 180);

        latitudeSegments  = new SegmentTree(ArrayUtils.toPrimitive(latitudeValues.toArray(new Double[latitudeValues.size()])));
        longitudeSegments = new SegmentTree(ArrayUtils.toPrimitive(longitudeValues.toArray(new Double[longitudeValues.size()])));
    }

    /**
     * Hyuga Event Processing Network (Set of EPAs) Definition.
     * 
     * @param scenario the domain values.
     * @param debug if debug rules should be used.
     */
    private void prepareCEP(Scenario scenario, boolean debug) {
        // Parameters
        double eps        = scenario.getEps();
        double timewindow = scenario.getTimeWindow();
        int    minPts     = scenario.getMinPts();
        
        // Configure the CEP Engine
        Configuration cepConfig = configureCEP();

        // CEP Engine
        cepEngine = EPServiceProviderManager.getProvider("ESPER", cepConfig);

        // CEP Rules
        EPAdministrator cepAdmin = cepEngine.getEPAdministrator();

        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
        // Named Windows
        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
        // String cellContextPartition = " CREATE CONTEXT Cell PARTITION BY x AND y FROM LocationUpdateEvent";
        String clusterNameWindow = " CREATE WINDOW  Clusters.win:keepall() AS GridLogicalClusterEvent";
        String debugNamedWindow  = " CREATE WINDOW  CC.win:time(" + (timewindow) + " sec).std:unique(x,y) AS CellContentEvent";
        
        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
        // Partition (Cell) Cluster
        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
        String cellContentEPA            = //" @Name('Cell Content EPA') @Audit"
                                           " INSERT INTO CellContentEvent "
                                         + " SELECT x, y, window(lu.*).where(p => p.x=lu.x AND p.y=lu.y AND p.valid=true) AS nodes"
                                         + " FROM   LocationUpdateEvent(valid = true).win:time(" + timewindow + " sec).std:unique(uuid) AS lu";

        String cellCoreClusterEPA        = " INSERT INTO CellCoreEvent "
                                         + " SELECT x, y, nodes"
                                         + " FROM   CellContentEvent AS cell"
                                         + " WHERE  cell.size() >= " + (minPts);

        String cellClusterDetectionEPA   = " INSERT INTO CellClusterEvent "
                                         + " SELECT core, "
                                         + "        (SELECT window(border.*)"
                                         + "         FROM   CellContentEvent.win:time(" + timewindow + " sec).std:unique(x,y) AS border "
                                         + "         WHERE  ("
                                         + "                 (Math.abs(border.x - core.x) <= 1) AND"
                                         + "                 (Math.abs(border.y - core.y) <= 1) AND"
                                         + "                 ((Math.abs(border.x - core.x) != 0) OR (Math.abs(border.y - core.y) != 0))" 
                                         + "                )) AS neighbors"
                                         + " FROM   CellCoreEvent AS core ";
        
        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
        // Transient (Cell Cluster)
        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
        String cellTransientEPA          = " INSERT INTO CellTransientEvent "
                                         + " SELECT x, y, nodes"
                                         + " FROM   CellContentEvent AS cell"
                                         + " WHERE  cell.size() < " + minPts + " AND cell.size() >=" + scenario.getCellMinMinPts();

        String cellTransientCheckEPA     = " INSERT INTO CellTransientCheckEvent "
                                         + " SELECT transientCell AS core, "
                                         + "        (SELECT window(border.*)"
                                         + "         FROM   CellContentEvent.win:time(" + timewindow + " sec).std:unique(x,y) AS border "
                                         + "         WHERE  ("
                                         + "                 (Math.abs(border.x - transientCell.x) <= 1) AND"
                                         + "                 (Math.abs(border.y - transientCell.y) <= 1) AND"
                                         + "                 ((Math.abs(border.x - transientCell.x) != 0) OR (Math.abs(border.y - transientCell.y) != 0))" 
                                         + "                )) AS neighbors"
                                         + " FROM   CellTransientEvent AS transientCell ";
        
        String cellTransientClusterEPA   = " INSERT INTO CellClusterEvent "
                                         + " SELECT toCellCluster(transientCell)"
                                         + " FROM   CellTransientCheckEvent AS transientCell"
                                         + " WHERE  isDense(" + eps + ", " + minPts + ", " + scenario.getCellSubDivision() + ","
                                         + "                transientCell.getCoreCellContent(), transientCell.getNeighbors())";

        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
        // Cell Dispersion
        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
        String cellClusterDispByTimeEPA  = " INSERT INTO CellDisperseEvent " 
                                         + " SELECT cellCluster.core.x AS x, cellCluster.core.y AS y, "
                                         + "        current_timestamp AS timestamp"
                                         + " FROM PATTERN "
                                         + "   [ " 
                                         + "     EVERY cellCluster=CellClusterEvent -> "
                                         + "     (timer:interval(" + 2 * timewindow + " sec) AND NOT"
                                         + "      CellClusterEvent(core.x=cellCluster.core.x, core.y=cellCluster.core.y))"
                                         + "   ] ";
        
        String cellLUChangedCellEPA      = " INSERT INTO CellChangedEvent " 
                                         + " SELECT * "
                                         + " FROM LocationUpdateEvent(valid=true).win:time(" + timewindow + " sec) "
                                         + " MATCH_RECOGNIZE ("
                                         + "   PARTITION BY uuid"
                                         + "   MEASURES A AS previous, B AS current "
                                         + "   AFTER MATCH SKIP TO NEXT ROW"
                                         + "   PATTERN (A B) "
                                         + "   DEFINE "
                                         + "     B AS B.x != A.x OR B.y != B.y " 
                                         + " )";

        String cellRemoveChangedLUEPA    = " INSERT INTO LocationUpdateEvent " 
                                         + " SELECT previous.uuid AS uuid, previous.originalID AS originalID,"
                                         + "        previous.icon AS icon, previous.x AS x, previous.y AS y,"
                                         + "        false AS valid"
                                         + " FROM CellChangedEvent.win:time(" + timewindow + " sec)";
        
        String cellRecheckNeededEPA      = " INSERT INTO CellRecheckNeededEvent "
                                         + " SELECT lu.x AS x, lu.y AS y"
                                         + " FROM PATTERN "
                                         + "   [ " 
                                         + "     EVERY cellCluster=CellClusterEvent -> "
                                         + "     EVERY lu=LocationUpdateEvent(valid=false,x=cellCluster.core.x, y=cellCluster.core.y)"
                                         + "     AND NOT CellClusterEvent(core.x=cellCluster.core.x, core.y=cellCluster.core.y)"
                                         + "     AND NOT CellDisperseEvent(x=cellCluster.core.x,y=cellCluster.core.y)"
                                         + "   ] ";
        
        String cellRecheckEPA            = " INSERT INTO CellRecheckEvent "
                                         + " SELECT "
                                         + "        (SELECT window(lu.*)"
                                         + "         FROM LocationUpdateEvent(valid=true).win:time(" + timewindow + " sec).std:unique(uuid) AS lu"
                                         + "         WHERE lu.x = cell.x AND lu.y = cell.y) AS rawCore,"
                                         + "        (SELECT window(border.*)"
                                         + "         FROM   CellContentEvent.win:time(" + timewindow + " sec).std:unique(x,y) AS border "
                                         + "         WHERE  ("
                                         + "                 (Math.abs(border.x - cell.x) <= 1) AND"
                                         + "                 (Math.abs(border.y - cell.y) <= 1) AND"
                                         + "                 ((Math.abs(border.x - cell.x) != 0) OR (Math.abs(border.y - cell.y) != 0))" 
                                         + "                )) AS neighbors"
                                         + " FROM CellRecheckNeededEvent AS cell";

        String cellDispersionByChangeEPA = " INSERT INTO CellDisperseEvent "
                                         + " SELECT cell.getX() AS x, cell.getY() AS y,"
                                         + "        current_timestamp AS timestamp"
                                         + " FROM  CellRecheckEvent AS cell"
                                         + " WHERE NOT isDense(" + eps + ", " + minPts + ", " + scenario.getCellSubDivision() + ","
                                         + "                   cell.getCoreCell(), cell.getNeighbors())";
        
        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
        // Grid Cluster
        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
        String gridClusterCheckEPA       = " ON CellClusterEvent(distributed = true) AS cellCluster"
                                         + " MERGE INTO Clusters AS candidateCluster"
                                         + " WHERE containsCell(cellCluster.core.x, cellCluster.core.y, candidateCluster)"
                                         + " WHEN MATCHED "
                                         + "      THEN INSERT INTO UpdateGridClusterEvent SELECT coreCells"
                                         + "      THEN UPDATE SET  timestamp = current_timestamp"
                                         + " WHEN NOT MATCHED "
                                         + "      THEN INSERT INTO Clusters            SELECT toGridCluster(cellCluster) "
                                         + "      THEN INSERT INTO NewCellClusterEvent SELECT cellCluster AS cell, "
                                         + "                                                  toGridCluster(cellCluster) AS cluster";
        
        String gridWriteModifyEPA        = " SELECT *"
                                         + " FROM NewCellClusterEvent";

        String gridMergeEPA              = " ON LockedCellClusterEvent as cellCluster"
                                         + " INSERT INTO MergeGridClusterEvent "
                                         + " SELECT AND DELETE mergeCore(cellCluster.cell, window(candidateCluster.*)) AS coreCells"
                                         + " FROM   Clusters as candidateCluster "
                                         + " WHERE  isContainedOrNeighbor(cellCluster.cell.core.x, cellCluster.cell.core.y, candidateCluster.coreCells)"
                                         + " LIMIT 1";
        
        String gridMergeUnlockEPA        = " INSERT INTO Clusters"
                                         + " SELECT coreCells"
                                         + " FROM   MergeGridClusterEvent ";
        
        String gridDispLockEPA           = " SELECT *"
                                         + " FROM CellDisperseEvent(distributed = true)";
        
        String gridDispExtractEPA        = " ON LockedCellDisperseEvent AS disperse"
                                         + " INSERT INTO DispersedGridClusterEvent"
                                         + " SELECT AND DELETE disperse.x AS x,"
                                         + "                   disperse.y AS y,"
                                         + "                   candidateCluster AS cluster"
                                         + " FROM Clusters AS candidateCluster"
                                         + " WHERE containsCell(x, y, candidateCluster)"
                                         + " LIMIT 1";
        
        String gridDispersionEPA         = " SELECT *"
                                         + " From DispersedGridClusterEvent";
        
        String gridDispUnlockEPA         = " SELECT *"
                                         + " FROM UnlockCellDisperseEvent";
        
        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
        // Grid Semantic
        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////        
        String gridAddEPA                = " SELECT newCluster "
                                         + " FROM   NewCellClusterEvent AS newCluster";
        
        String gridUpdateEPA             = " SELECT upCluster, "
                                         + "        (SELECT window(cell.*)"
                                         + "         FROM   CellContentEvent.std:unique(x,y).win:time(" + timewindow + " sec) AS cell "
                                         + "         WHERE  isContainedOrNeighbor(cell.x, cell.y, upCluster.coreCells)) AS cells"
                                         + " FROM   UpdateGridClusterEvent AS upCluster";
        
        String gridMergeOutputEPA        = " SELECT mergeCluster, "
                                         + "        (SELECT window(cell.*)"
                                         + "         FROM   CellContentEvent.std:unique(x,y).win:time(" + timewindow + " sec) AS cell "
                                         + "         WHERE  isContainedOrNeighbor(cell.x, cell.y, mergeCluster.coreCells)) AS cells"
                                         + " FROM   MergeGridClusterEvent AS mergeCluster";
 
        String gridDispOutputEPA         = " SELECT *"
                                         + " From DispersedGridClusterEvent";

       
        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
        // Output EPAS and DEBUG EPAs
        ///////////////////////////////////////////////////////////////////////////////////////////////////////////////
        String csvOutEPA                 = " SELECT * "
                                         + " FROM CSVLocationUpdateEvent.win:time(" + timewindow + " sec)";
        
        String cellClusterFormOutEPA     = " SELECT * "
                                         + " FROM CellClusterEvent(distributed=false)";
        
        String cellClusterDispOutEPA     = " SELECT * "
                                         + " FROM CellDisperseEvent(distributed=false)";
        
        String debugEPA                  = " INSERT INTO CC "
                                         + " SELECT *"
                                         + " FROM   CellContentEvent";
        
        // We create and deploy basic stream rules (Context and Windows)
        // cepAdmin.createEPL(cellContextPartition);
        epls.put(EPA.GRIDWINDOW, new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(clusterNameWindow), null));
        deploy(EPA.GRIDWINDOW);
        
        // Cell EPA
        epls.put(EPA.CELLCHECK,      new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellContentEPA),           null));
        epls.put(EPA.CELLCORE,       new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellCoreClusterEPA),       null));
        epls.put(EPA.CELLCLUSTER,    new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellClusterDetectionEPA),  null));
        
        // Cell Dispersion
        epls.put(EPA.CELLDISPERSIONBYTIME,   new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellClusterDispByTimeEPA),  null));
        epls.put(EPA.CELLCHANGED,            new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellLUChangedCellEPA),      null));
        epls.put(EPA.CELLREMOVECHANGEDLU,    new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellRemoveChangedLUEPA),    null));
        epls.put(EPA.CELLRECHECKTEST,        new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellRecheckNeededEPA),      null));
        epls.put(EPA.CELLRECHECK,            new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellRecheckEPA),            null));
        epls.put(EPA.CELLDISPERSIONBYCHANGE, new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellDispersionByChangeEPA), null));

        if (scenario.isUsingCellSubDiv()) {
            epls.put(EPA.CELLTRANSIENT,        new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellTransientEPA),        null));
            epls.put(EPA.CELLTRANSIENTCHECK,   new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellTransientCheckEPA),   null));
            epls.put(EPA.CELLTRANSIENTCLUSTER, new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellTransientClusterEPA), null));
        }

        // Grid Cluster EPA
        GridLockListener gridLockListener = new GridLockListener(this);
        epls.put(EPA.GRIDMERGECHECK,  new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(gridClusterCheckEPA), new GridReadUnlockListener()));
        epls.put(EPA.GRIDMERGELOCK,   new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(gridWriteModifyEPA),  gridLockListener));
        epls.put(EPA.GRIDMERGE,       new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(gridMergeEPA),        null));
        epls.put(EPA.GRIDMERGEUNLOCK, new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(gridMergeUnlockEPA),  null));
        
        // Grid Disp EPA
        epls.put(EPA.GRIDDISPLOCK,    new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(gridDispLockEPA),    gridLockListener));
        epls.put(EPA.GRIDDISPEXTRACT, new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(gridDispExtractEPA), null));
        epls.put(EPA.GRIDDISPERSION,  new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(gridDispersionEPA),  new ClusterDispersionListener(dataStream, cepEngine)));
//        epls.put(EPA.GRIDDISPUNLOCK,  new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(gridDispUnlockEPA),  new GridWriteUnlockListener()));

        // Semantic
        epls.put(EPA.GRIDADD,         new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(gridAddEPA),         output));
        epls.put(EPA.GRIDUPDATE,      new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(gridUpdateEPA),      output));
        epls.put(EPA.GRIDMERGEOUTPUT, new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(gridMergeOutputEPA), output));
        epls.put(EPA.GRIDDISPOUTPUT,  new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(gridDispOutputEPA),  output));
        
        // Output EPAs
        CellOutputListener cellOutputListener = new CellOutputListener(this, dataStream);
        epls.put(EPA.CSVSTREAM,          new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(csvOutEPA),             new CSVListener(dataStream)));
        epls.put(EPA.CELLCLUSTERFORMOUT, new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellClusterFormOutEPA), cellOutputListener));
        epls.put(EPA.CELLCLUSTERDISPOUT, new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(cellClusterDispOutEPA), cellOutputListener));
        
        if (debug) {
            cepAdmin.createEPL(debugNamedWindow);
            epls.put(EPA.DEBUG, new MutablePair<EPPreparedStatement, Object>(cepAdmin.prepareEPL(debugEPA), null));
        }
    }
  
    private Configuration configureCEP() {
        Configuration cepConfig = new Configuration();

        // Threading and Tuning
        Threading threading = cepConfig.getEngineDefaults().getThreading();
        threading.setThreadPoolInbound(false);
        // threading.setThreadPoolInboundNumThreads(4);
        threading.setThreadPoolOutbound(false);
        // threading.setThreadPoolOutboundNumThreads(16);
        threading.setInsertIntoDispatchPreserveOrder(false);
        threading.setListenerDispatchPreserveOrder(false);
        threading.setInsertIntoDispatchTimeout(250);
        cepConfig.getEngineDefaults().getViewResources().setShareViews(false);
        
        // Event Types - Location Updates
        cepConfig.addEventType(LocationUpdateEvent.class);
        cepConfig.addEventType(CSVLocationUpdateEvent.class);

        // Event Types - Cell
        cepConfig.addEventType(CellClusterEvent.class);
        cepConfig.addEventType(CellContentEvent.class);
        cepConfig.addEventType(CellCoreEvent.class);
        // Event Types - Cell Dispersion
        cepConfig.addEventType(CellDisperseEvent.class);
        cepConfig.addEventType(CellChangedEvent.class);
        cepConfig.addEventType(CellRecheckNeededEvent.class);
        cepConfig.addEventType(CellRecheckEvent.class);
        // Event Types - Cell (Blind Spot)
        cepConfig.addEventType(CellTransientEvent.class);
        cepConfig.addEventType(CellTransientCheckEvent.class);
        
        // Event Types - Grid
        cepConfig.addEventType(GridClusterEvent.class);
        cepConfig.addEventType(GridLogicalClusterEvent.class);
        cepConfig.addEventType(LockedCellClusterEvent.class);
        cepConfig.addEventType(LockedCellDisperseEvent.class);
        cepConfig.addEventType(UnlockCellDisperseEvent.class);
        // Event Types - Grid Semantic
        cepConfig.addEventType(NewCellClusterEvent.class);
        cepConfig.addEventType(NewCellDisperseEvent.class);
        cepConfig.addEventType(UpdateGridClusterEvent.class);
        cepConfig.addEventType(MergeGridClusterEvent.class);
        cepConfig.addEventType(DispersedGridClusterEvent.class);

        ////////////////////////////////////////////////////////////
        // Functions
        ////////////////////////////////////////////////////////////
        
        // Neighboring Functions
        cepConfig.addPlugInSingleRowFunction("isNeighbor",            Grid.class.getCanonicalName(), "isNeighbor");
        cepConfig.addPlugInSingleRowFunction("containsCell",          Grid.class.getCanonicalName(), "containsCell");
        cepConfig.addPlugInSingleRowFunction("isContainedOrNeighbor", Grid.class.getCanonicalName(), "isContainedOrNeighbor");

        // Mapping/Casting Function
        cepConfig.addPlugInSingleRowFunction("toGridCluster", Grid.class.getCanonicalName(), "toGridCluster");
        
        // Decay Functions
        cepConfig.addPlugInSingleRowFunction("isDense",       Grid.class.getCanonicalName(), "isDense");
        cepConfig.addPlugInSingleRowFunction("toCellCluster", Grid.class.getCanonicalName(), "toCellCluster");

        // Lock Functions
        cepConfig.addPlugInSingleRowFunction("gridWriteLock",      Grid.class.getCanonicalName(), "gridWriteLock");
        cepConfig.addPlugInSingleRowFunction("gridWriteUnlock",    Grid.class.getCanonicalName(), "gridWriteUnlock");
       
        // Grid Functions
        cepConfig.addPlugInSingleRowFunction("mergeCore",     Grid.class.getCanonicalName(), "mergeCore");
        
        // Debug
        cepConfig.addPlugInSingleRowFunction("debug",         Grid.class.getCanonicalName(), "debug");
        
        return cepConfig;
    }
    
    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    // Deploy/UnDeploy API
    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

    
    /**
     * Deploys the EPAs of the given EPA set.
     * 
     * @param epaSet a Set containing the EPAs to be deployed.
     */
    public void deploy(EnumSet<EPA> epaSet) {
        for (EPA epa : epaSet) {
            deploy(epa);
        }
    }
    
    /**
     * Deploy a specific EPA
     * 
     * @param epa the EPA to be deployed
     * @return the EPA statement
     */
    public EPStatement deploy(EPA epa) {
        if (epls.get(epa) != null)
            return deploy(epa, epls.get(epa).getRight());
        else
            return null;
    }
    
    /**
     * Deploys an EPA with a given listener.
     * 
     * @param epaName the EPA to be deployed
     * @param listener the EPA output listener
     * @return
     */
    public EPStatement deploy(EPA epaName, Object listener) {
        if (deployedEPAs.containsKey(epaName) && !deployedEPAs.get(epaName).isDestroyed()) {
            LOGGER.info("EPA " + epaName + " has alread been deployed.");
            
            return deployedEPAs.get(epaName);
        }
        
        Pair<EPPreparedStatement, Object> preparedEPA = epls.get(epaName);
        EPStatement stmt = cepEngine.getEPAdministrator().create(preparedEPA.getLeft(), epaName.toString());
        
        if (listener != null) {
            if (listener instanceof UpdateListener)
                stmt.addListener((UpdateListener) listener);
            else
                stmt.setSubscriber(listener);
        }
        
        deployedEPAs.put(epaName, stmt);
        LOGGER.info("Deployed EPA: " + epaName);
        return stmt;
    }
    
    /**
     * Undeploy the EPAs of the given EPA set.
     * 
     * @param epaSet a Set containing the EPAs to be undeployed (destroyed).
     */
    public void undeploy(EnumSet<EPA> epaSet) {
        for (EPA epa : epaSet) {
            undeploy(epa);
        }
    }

    /**
     * Undeploys a specific EPA.
     * 
     * @param epaName the EPA to be undeployed (destroyed)
     * @return true if successful, false otherwise
     */
    public boolean undeploy(EPA epaName) {
        if (deployedEPAs.containsKey(epaName)) {
            deployedEPAs.get(epaName).destroy();
            LOGGER.info("Undeployed EPA: " + epaName + " successfully.");
            
            deployedEPAs.remove(epaName);
            return true;
        } else {
            LOGGER.error("Error at undeploying epa : " + epaName + ". " + epaName + " not found.");
            return false;
        }
    }
    
    public void resetEPAs() {
        for (EPA epa : deployedEPAs.keySet()) {
            undeploy(epa);
        }
    }
    
    public void resetWindow() {
        undeploy(EPA.GRIDWINDOW);
        deploy(EPA.GRIDWINDOW);
    }
    
    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    // Input Handling
    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

    /** 
     * Stream Receiver Function
     * It receives a location update event. This event is enriched with the context partition (pair x, y) related to the
     * event coordinates (latitude and longitude). Then, the event is injected in the Hyuga EPN.
     * 
     * @param dataStreamMsg a Spatial Message containing a CEP event.
     */
    @Override
    public void onNewData(ApplicationObject dataStreamMsg) {
        // We can apply filter hooks here

        SpatialMessage spatialMSG = (SpatialMessage) dataStreamMsg;
        recvPool.submit(new ReceiveMessage(spatialMSG));
    }
    
    public void onNewData(CEPEvent cepEvent) {
        cepEngine.getEPRuntime().sendEvent(cepEvent);
    }
    
    private class ReceiveMessage implements Runnable {
        private SpatialMessage message;
        
        public ReceiveMessage(SpatialMessage message) {
            this.message = message;
        }
        
        @Override
        public void run() {
            // Received MSG Type (default is location update)
            CEPEvent.TYPE type   = TYPE.LOCATIONUPDATE;
            EPRuntime cepRuntime = cepEngine.getEPRuntime();
            
            // Content type
            try {
                type = CEPEvent.TYPE.valueOf(message.getMsgType());                
            } catch (Exception ex) {
                LOGGER.warn("Error at detecting received event type. Assume that is a location update.");
                type = TYPE.LOCATIONUPDATE; 
            }
            
            switch (type) {
                case CELLCLUSTER:
                    try {
                        String cellClusterJSON = new String(message.getContent());
                        CellClusterEvent cellClusterEvent = jsonMapper.readValue(cellClusterJSON, CellClusterEvent.class);
                        int x = cellClusterEvent.getCore().getX();
                        int y = cellClusterEvent.getCore().getY();
                        
                        Grid.cellLock(x, y);
                        Grid.gridReadLock();

                        // Unwrapping cell cluster event content
                        cepRuntime.sendEvent(cellClusterEvent.getCore());
                        for (CellContentEvent cc : cellClusterEvent.toCellContentArray()) {
                            cepRuntime.sendEvent(cc);
                        }

                        cepRuntime.sendEvent(cellClusterEvent);
                        
                        Grid.cellUnlock(x, y);
                    } catch (IOException ex) {
                        LOGGER.error("Error at unwrapping CellCluster Event.");
                    }
                    break;
                case CELLDISPERSE:
                    try {
                        String cellDisperseJSON = new String(message.getContent());
                        CellDisperseEvent cellClusterEvent = jsonMapper.readValue(cellDisperseJSON, CellDisperseEvent.class);
                        cepRuntime.sendEvent(cellClusterEvent);            
                    } catch (IOException ex) {
                        LOGGER.error("Error at unwrapping CellDisperse Event.");
                    }
                    break;
                case LOCATIONUPDATE:
                    try {
                        String locationUpdateJSON = new String(message.getContent());
                        LocationUpdateEvent locationUpdateEvent = jsonMapper.readValue(locationUpdateJSON, LocationUpdateEvent.class);
                        cepRuntime.sendEvent(handleLocationUpdate(locationUpdateEvent));
                    } catch (IOException ex) {
                        LOGGER.error("Error at unwrapping LocationUpdate Event.");
                    }
                    break;
                default:
                    LOGGER.error("Error at unwrapping received event.");
                    break;
            }
        }
    }
    
    private CEPEvent handleLocationUpdate(LocationUpdateEvent locationUpdateEvent) {
        // Mapping to a Cell
        int x = longitudeSegments.findIndex(locationUpdateEvent.getLongitude());
        int y = latitudeSegments.findIndex(locationUpdateEvent.getLatitude());
        locationUpdateEvent.setX(x);
        locationUpdateEvent.setY(y);
    
        // Mapping to inner Cell
        double[] lngSegments = longitudeSegments.getSegments();
        double[] latSegments = latitudeSegments.getSegments();
        
        double lngSubSegSize = (lngSegments[x + 1] - lngSegments[x]) / scenario.getCellSubDivision();
        double latSubSegSize = (latSegments[y + 1] - latSegments[y]) / scenario.getCellSubDivision();

        if (scenario.isUsingCellSubDiv()) {
            int hsweep = (int) Math.min(getScenario().getCellSubDivision() - 1,
                                       ((locationUpdateEvent.getLongitude() - lngSegments[x]) / lngSubSegSize));
            int vsweep = (int) Math.min(getScenario().getCellSubDivision() - 1,
                                       ((locationUpdateEvent.getLatitude()  - latSegments[y])  / latSubSegSize));

            locationUpdateEvent.setHsweep(hsweep);
            locationUpdateEvent.setVsweep(vsweep);
        }
        
        return locationUpdateEvent;
    }
    
    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    // Get and Sets
    ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    
    public Scenario getScenario() {
        return scenario;
    }

    public SegmentTree getLatitudeSegments() {
        return latitudeSegments;
    }

    public SegmentTree getLongitudeSegments() {
        return longitudeSegments;
    }

    public EPServiceProvider getCEPEngine() {
        return cepEngine;
    }
    
    public boolean isLocalDeploy() {
        return localDeploy;
    }

    public void setLocalDeploy(boolean localDeploy) {
        this.localDeploy = localDeploy;
    }
}
